package org.qiyu.live.user.provider.config;


import jakarta.annotation.Resource;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MQProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.*;


@Configuration
public class RocketMQProducerConfig {


    private final static Logger LOGGER = LoggerFactory.getLogger(RocketMQProducerConfig.class);

    @Resource
    private RocketMQProducerProperties producerProperties;


    @Value("spring.application.name")
    private String appName;

    @Bean
    public MQProducer mqProducer(){
        ThreadPoolExecutor ansyThreadPoolExecutor = new ThreadPoolExecutor(100, 150, 3, TimeUnit.MINUTES, new ArrayBlockingQueue<>(1000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName(appName + "qiyu-rmq"+ThreadLocalRandom.current().nextInt(1000));
                return thread;
            }
        });

        DefaultMQProducer defaultMQProducer = new DefaultMQProducer();

        try {
            defaultMQProducer.setProducerGroup(producerProperties.getGroupName());
            defaultMQProducer.setNamesrvAddr(producerProperties.getNameSrv());
            defaultMQProducer.setRetryTimesWhenSendFailed(producerProperties.getRetryTimes());
            defaultMQProducer.setRetryTimesWhenSendAsyncFailed(producerProperties.getRetryTimes());
            defaultMQProducer.setRetryAnotherBrokerWhenNotStoreOK(true);
            defaultMQProducer.setAsyncSenderExecutor(ansyThreadPoolExecutor);
            defaultMQProducer.start();

            LOGGER.info("mq生产者启动成功,nameSrv is {}", producerProperties.getNameSrv());
        } catch (MQClientException e) {
            throw new RuntimeException(e);
        }

        return defaultMQProducer;

    }


}
